/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package edu.isi.pegasus.planner.subworkflow.scheduler;


import edu.isi.pegasus.planner.subworkflow.partitioner.SiteAbstraction;
import edu.isi.pegasus.planner.subworkflow.partitioner.PartitionerImplementation;
import edu.isi.pegasus.planner.subworkflow.partitioner.DAXAbstraction;
import edu.isi.pegasus.common.util.DynamicLoader;
import edu.isi.pegasus.planner.catalog.site.classes.SiteCatalogEntry;
import edu.isi.pegasus.planner.catalog.site.classes.SiteStore;
import edu.isi.pegasus.planner.classes.ADag;
import edu.isi.pegasus.planner.classes.PCRelation;
import edu.isi.pegasus.planner.classes.PegasusBag;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Vector;
/**
 *
 * @author wchen
 * @weather  66F, Tue Apr 12,15:58
 */


public abstract class PureScheduler implements SchedulerImplementation{

    protected ADag mDag ;
    //private Map mJob2DAX;
    protected ArrayList mParentDAX;
    protected ArrayList mChildDAX;
    private Map mDaxRC;

    protected Map mDaxMap;
    protected PegasusBag mBag;

    protected ArrayList mSiteIndex;
    private String reorder;

    //remove at first and then add
    private List<PCRelation> mReducedMapDag;
    //add at first and then remove
    private List<PCRelation> mReducedMapDag2;
    protected LinkedList mDaxQueue;
    protected Map mJob2Time;
    protected Map mDAXID2DAX;
    public static final String DEFAULT_PARTITIONER_PATH =
        "edu.isi.pegasus.planner.subworkflow.partitioner.implementation";
    public static final String DEFAULT_ESTIMATOR_PATH =
         "edu.isi.pegasus.planner.subworkflow.estimator.implementation";
    public static final String DEFAULT_PARTITIONER = "BackwardPartitioner8";
    public static final String DEFAULT_ESTIMATOR = "CriticalPath";
    public Map getDAXRC()
    {
        return mDaxRC;
    }
    public PureScheduler(ADag dag, PegasusBag bag){
        mDag = dag;
        mBag = bag;
        //mJob2DAX = new HashMap<String, ADAG>();
        mParentDAX = new ArrayList<String>();
        mChildDAX = new ArrayList<String>();
        //mDaxRC =  new HashMap<String, ReplicaLocation>();
        //mDaxSize = new HashMap<String, Double>();
        //mDaxList = new ArrayList<ADAG>();
        mDaxMap = new HashMap<DAXAbstraction, Map>();
        //mSiteMap =  new HashMap<Integer, String>();
        mReducedMapDag = new ArrayList<PCRelation>();
        mReducedMapDag2 = new ArrayList<PCRelation>();
        mSiteIndex = new ArrayList<SiteAbstraction>();
        //mSiteSizeMap = new HashMap<Integer, Double>();
        mDaxQueue = new LinkedList<DAXAbstraction>();
        mJob2Time = new HashMap();
        mDAXID2DAX = new HashMap<String, DAXAbstraction>();

    }

    private void reduceByLabel()
    {
        Vector relation = mDag.dagInfo.relations;
        int sum = 0;
        int l2size = relation.size();
        PCRelation pcr = new PCRelation();
        PCRelation bcr = null;
        PCRelation tcr = null;
        boolean hasAdd = false;
        for(int i = 0; i < l2size; i ++)
         {

            PCRelation rel = (PCRelation)relation.get(i - sum);


            if( (rel.parent.contains("mProject") && rel.child.contains("mBackground"))
                    || (rel.parent.contains("mBackground") && rel.child.contains("mAdd"))
                    || (rel.parent.contains("mShrink") && rel.child.contains("mAdd"))
                    || (rel.parent.contains("mBackground") && rel.child.contains("mAdd")))
                    {
                        mReducedMapDag.add(rel);
                        relation.remove(rel);
                        //System.out.println(rel);
                        sum ++;
                    }
            //For CyberShake
            if(rel.child.contains("_ZipSeis") )
            {
                pcr.setChild(rel.child);
                mReducedMapDag.add(rel);
                relation.remove(rel);
                hasAdd = true;
                sum ++;
            }
            if(rel.parent.contains("ZipPSA") || rel.parent.contains("ZipPeakSA"))
            {
                pcr.setParent(rel.parent);
                bcr = rel;
                relation.remove(bcr);
                sum ++;
            }
            if(rel.parent.contains("_ZipSeis"))
            {
                tcr = rel;
                relation.remove(tcr);
                sum ++;
            }

        }

        if(hasAdd)
        {
            relation.add(pcr);
            //relation.add(bcr);
            relation.add(tcr);
            mReducedMapDag2.add(pcr);
        }



    }
    private void restore()
    {
        for(Iterator it = mReducedMapDag.iterator(); it.hasNext();)
        {
            PCRelation rel =(PCRelation)it.next();
            mDag.dagInfo.relations.add(rel);
        }
        for(Iterator it = mReducedMapDag2.iterator(); it.hasNext();)
        {
            PCRelation rel =(PCRelation)it.next();
            mDag.dagInfo.relations.remove(rel);
        }


    }
    public void addDAXJob()
    {
        String reduceMethod = mBag.getPegasusProperties().getProperty("pegasus.partition.reduce");

        if(reduceMethod!=null && reduceMethod.equals("label"))
            reduceByLabel();

        // The usual way of a dax name is subworkflow0, subworkflow1
        getProperties();
        PartitionerImplementation p ;

        String className = this.mBag.getPegasusProperties().getProperty("pegasus.subworkflow.partitioner");
        if(className == null || className.equals(""))
            className = DEFAULT_PARTITIONER;

        System.out.println("Selected Paritioner is " + className);
        className = (className.indexOf('.') == -1) ? DEFAULT_PARTITIONER_PATH + "." + className : className;
        DynamicLoader dl_pa = new DynamicLoader(className);
        //System.out.println("ClassName:" + className);
        Object argList[] = new Object[2];
        argList[0] = mDag;
        argList[1] = mBag;



        try
        {
           p = (PartitionerImplementation) dl_pa.instantiate(argList);
           p.run();
 //           PartitionResult pr = p.getResult();
//            mDaxMap = pr.mDaxMap;
//            mDaxRC = pr.mDaxRC;//this is different
//            mJob2Time = pr.mJob2Time;
           restore();
           mDag = p.getDag();//make it parallel

      

        }
        catch (Exception ex)
        {
            ex.printStackTrace();
        }

    }

    public void siteSelector()
    {
        System.out.println("You're wrong to be here\n");
    }

   
    private void getProperties()
    {
        try
        {

            SiteStore mSite = mBag.getHandleToSiteStore();

            for (Iterator it = mSite.entryIterator();it.hasNext();)
            {
                SiteCatalogEntry siteEntry = (SiteCatalogEntry)it.next();
                String siteName = siteEntry.getSiteHandle();
                if(siteName.equals("local"))continue;
                double size = Double.parseDouble(siteEntry.getEnvironmentVariable("SITESIZE")) * 1e8;

                int slot    = Integer.parseInt(siteEntry.getEnvironmentVariable("SLOT"));

                SiteAbstraction site = new SiteAbstraction(siteName, size, slot);
                mSiteIndex.add(site);
                System.out.println(siteName+" has been selected in SubworkflowPartitioner.java");

            }
            reorder = mBag.getPegasusProperties().getProperty("pegasus.subworkflow.reorder");

        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
    }
}
